package com.techstar.ats.netty;

import java.util.Optional;
import java.util.concurrent.atomic.AtomicReference;
import lombok.extern.slf4j.Slf4j;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import com.google.common.util.concurrent.AbstractService;
import com.techstar.ats.ConnectProperties;
import com.techstar.ats.netty.handler.*;
import com.techstar.ats.utils.SharedUtils;

@Slf4j
public class AtsNettyClient extends AbstractService {
	private final ConnectProperties properties;
	protected final Bootstrap bootstrap;
	protected final AtomicReference<Channel> channelRef = new AtomicReference<>();

	public AtsNettyClient(ConnectProperties properties) {
		this.properties = properties;
		this.bootstrap = new Bootstrap()
			.group(SharedUtils.sharedEventLoop())
			.channel(NioSocketChannel.class)
			.handler(channelInitializer())
			.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, properties.getConnectTimeout() * 1000)
			.remoteAddress(properties.toSocketAddress());
	}

	@Override
	protected void doStart() {
		open().addListener((ChannelFutureListener) future -> {
			if(future.isSuccess()) {
				channelRef.set(future.channel());
				notifyStarted();
			} else {
				notifyFailed(future.cause());
			}
		});
	}

	@Override
	protected void doStop() {
		Optional.ofNullable(channelRef.get())
			.ifPresent(channel -> channel.close()
				.addListener((ChannelFutureListener) future -> {
					if(future.isSuccess()) {
						notifyStopped();
					} else {
						notifyFailed(future.cause());
					}
					channelRef.set(null);
				}));
	}

	public Optional<Channel> getChannel() {
		return Optional.ofNullable(channelRef.get());
	}

	protected ChannelFuture open() {
		log.info("connecting to: {}", properties.toSocketAddress());
		return bootstrap.connect();
	}

	protected ChannelInitializer<Channel> channelInitializer() {
		return new ChannelInitializer<Channel>() {

			protected void initChannel(Channel ch) {
				ChannelPipeline pipeline = ch.pipeline();
				pipeline.addLast(new LoggingHandler());
				pipeline.addLast(new IdleStateHandler(
					properties.getReadTimeout(),
					properties.getWriteTimeout(),
					properties.getBothTimeout())
				);
				pipeline.addLast(new AtsDecoder());
				pipeline.addLast(new AtsResponseDecoder());
				pipeline.addLast(new AtsRequestEncoder());
				pipeline.addLast(new LinkLayerHandler(properties));
				pipeline.addLast(new KeepAliveHandler());
				pipeline.addLast(new CsvExportHandler());
				customHandler(pipeline);
				pipeline.addLast(new ChannelInboundHandlerAdapter() {
					@Override
					public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
						try {
							AtsNettyClient.this.stopAsync();
						} finally {
							super.channelUnregistered(ctx);
						}
					}
				});
			}
		};
	}

	protected void customHandler(ChannelPipeline pipeline) {
	}
}
